package com.linkstec.kafka;


import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 * 根据优先级消费数据
 * 示例 TEST_1>TEST_2>TEST_2......
 * @author linkage
 *
 */
public class KafkaConsumerRouterStartPriority {
    private static final Logger logger = LogManager.getLogger();

    //跟据topic分区数配置 = Partitions*n
    private static int PRIORITY_CONSUMERS_NUM;

    public void setRouter() {

        final List<KafkaRouterPriority> consumers = new ArrayList<KafkaRouterPriority>();

        String consumersNum = "1";//ConfigUtil.getProperty("priority.consumers.num");
        if (StringUtils.isEmpty(consumersNum)) {
            throw new RuntimeException("priority.consumers.num未配置，推送服务退出运行");
        }
        PRIORITY_CONSUMERS_NUM = Integer.parseInt(consumersNum);

        final ExecutorService executor = Executors.newFixedThreadPool(PRIORITY_CONSUMERS_NUM);
        logger.info("线程池大小:" + PRIORITY_CONSUMERS_NUM);
        for (int i = 0; i < PRIORITY_CONSUMERS_NUM; i++) {
            KafkaRouterPriority consumer = new KafkaRouterPriority();
            consumer.init(KafkaRouterPriority.PRIORITY_ROUTER_GROUPID,KafkaRouterPriority.PRIORITY_MESSAGE_TOPIC);
            consumers.add(consumer);
            executor.submit(consumer);
        }
        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                for (KafkaRouterPriority consumer : consumers) {
                    consumer.shutdown();
                }
                executor.shutdown();
                try {
                    executor.awaitTermination(5000, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }
    public static void main(String[] args) {
    	KafkaConsumerRouterStartPriority sss = new KafkaConsumerRouterStartPriority();
    	sss.setRouter();
	}
}